package com.bangcommunity.bbframe.mq.kafka.producer;

import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import com.bangcommunity.bbframe.common.utils.io.PropertiesUtil;
import com.bangcommunity.bbframe.common.utils.lang.SpiUtils;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.dynconfig.DynConfigRegister;
import com.bangcommunity.bbframe.dynconfig.IDynConfigProcessor;
import com.bangcommunity.bbframe.mq.IQtMessager;
import com.bangcommunity.bbframe.mq.producer.IQtProducer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class KafkaProducerClient {
    static Logger logger = LoggerFactory.getLogger(KafkaProducerClient.class);

    private KafkaProducerClient() {
    }

    private static class KafkaProducerClientHolder {
        private static DelegateKafkaProducer instance;
        private static final String producer_config_group = "common";
        private static final String producer_config_dataId = "kafkaProducer";
        private static final String productCode = MainConfigLoader.getInstance().getProductCode();
        private static final String appCode = MainConfigLoader.getInstance().getAppCode();
        static {
            DynConfigRegister.getInstance().buildOnce(productCode, appCode, producer_config_group,
                    producer_config_dataId, new IDynConfigProcessor() {
                        @Override
                        public void procesConfig(String config) {
                            Properties properties = PropertiesUtil.loadFromString(config);
                            DelegateKafkaProducer delegateKafkaProducer = buildDelegateKafkaProducer(properties);
                            if (null != delegateKafkaProducer) {
                                instance = delegateKafkaProducer;
                            } else {
                                logger.warn("didnot modify kafka build client is null:productCode={},appCode={}",
                                        productCode, appCode);
                            }
                        }
                    });
        }

        /**
         * 属性基于0.10.1
         */
        private static DelegateKafkaProducer buildDelegateKafkaProducer(Properties properties) {
            if (null == properties) {
                properties = new Properties();
            }
            // host/port对的列表，用来建立与kafka的初始链接。host1:port1,host2,port2,....
            // setDefault(properties,"bootstrap.servers","");
            // 更新metadata的时间间隔，无论partition的leader是否发生变换或者topic其它的元数据是否发生变化。
            // setDefault(properties,"metadata.max.age.ms","");//defalult: 300000
            // producer尝试批量处理消息，可以使用较少的发送次数发送相同数量的消息。
            // 可以提高server以及client端的性能。此值是指默认批量处理的字节数。
            // setDefault(properties,"batch.size","");//defalult: 16384
            // producer要求在leader在判定某条消息是否写入完成之前需要收到的确认写入的个数
            // setDefault(properties,"acks","all");//defalult:1
            // 该设置通过添加人为的延迟实现这一点：即，请求来到不是立即发送，而是等待指定的延迟，批量发送请求。
            // setDefault(properties,"linger.ms","");//defalult:0
            // 请求server时传送给server的clientid字符串。
            setDefault(properties, "client.id", productCode + "_" + appCode);// defalult:""
            // TCP发送的socket的SO_SNDBUF缓存。如果设置为－1，将使用OS的默认值
            // setDefault(properties,"send.buffer.bytes","");//defalult:131072
            // TCP接受缓存的大小（SO_RCVBUF）。如果设置为－1，则使用OS默认值
            // setDefault(properties,"receive.buffer.bytes","");//defalult:32768
            // 请求的最大字节数
            // setDefault(properties,"max.request.size","");//defalult:1048576
            // 重连给定host之前的等待时间。避免频繁的重连某个host。
            // 这个backoff时间也设定了consumer请求broker的重试等待时间。
            // setDefault(properties,"reconnect.backoff.ms","");//defalult:50
            // KafkaProducer.send（）和KafkaProducer.partitionsFor（）将会阻塞的时长
            // setDefault(properties,"max.block.ms","");//defalult:33554432
            // producer用于缓存发送数据的内存大小。如果消息放入缓存的速度大于发送的速度，
            // 则producer可以设置阻塞超时时间max.block.ms，超时则报异常即可
            // setDefault(properties,"buffer.memory","");//defalult:33554432
            // 重新发送失败请求的等待时间。避免某些失败情况下频繁发送请求。
            // setDefault(properties,"retry.backoff.ms","");//defalult:100
            // producer可以支持的数据压缩类型。合法的压缩格式为：none，gzip，snappy，lz4.
            // setDefault(properties, "compression.type", "");//defalult:none
            // 度量样本的计算的时长
            // setDefault(properties,"metrics.sample.window.ms","");//defalult:30000
            // 维护计算指标的样本数
            // setDefault(properties,"metrics.num.samples","");//defalult:2
            // 用于实现指标统计的类的列表。
            // setDefault(properties,"metric.reporters","");//defalult:[]
            // 在阻塞之前，客户端可以在单个链接之中发送未确认的请求的最大数目。
            // 注意：如果这个值大于1，则一旦发送失败，有可能会打乱消息的原有顺序
            // setDefault(properties, "max.in.flight.requests.per.connection", "");//defalult:5
            // 设置重试次数可以在发送失败时进行重试，提高发送的可靠性。
            // setDefault(properties, "retries", "");//defalult:0
            // 空闲链接存在的最长时间，超出之后就会被关闭
            // setDefault(properties,"connections.max.idle.ms","100");//defalult:540000
            // 实现Partitioner的接口的Partitioner类
            // setDefault(properties,"partitioner.class","");//defalult:
            // 请求的超时时间，即等待server端应答某个请求的最长时间
            // setDefault(properties, "request.timeout.ms", "");//defalult:30000
            /**
             * 以下参数不允许配置
             */
            // Serializer接口的密钥的类的key
            overrideValue(properties, "key.serializer",
                    "com.bangcommunity.bbframe.mq.kafka.serial.QtMessageSerializer");// defalult:
            // Serializer接口的类的value
            overrideValue(properties, "value.serializer",
                    "com.bangcommunity.bbframe.mq.kafka.serial.QtMessageSerializer");// defalult:
            List<ProducerInterceptor> producerInterceptors = SpiUtils.listSpi(ProducerInterceptor.class);
            StringBuffer intecptStr = new StringBuffer();
            if (null != producerInterceptors) {
                for (int i = 0; i < producerInterceptors.size(); i++) {
                    if (i != 0) {
                        intecptStr.append(",");
                    }
                    intecptStr.append(producerInterceptors.get(i).getClass().getName());
                }
            }
            // 用作拦截器的类的列表。
            overrideValue(properties, "interceptor.classes", intecptStr.toString());// defalult:
            return new DelegateKafkaProducer(new KafkaProducer<IQtMessager, IQtMessager>(properties));
        }

        private static void setDefault(Properties properties, String key, Object defalt) {
            if (StringUtils.isNotBlank(key) && null != properties && null != defalt) {
                Object o = properties.get(key);
                if (null == o) {
                    o = defalt;
                }
                properties.put(key, o);
            }
        }

        private static void overrideValue(Properties properties, String key, Object newValue) {
            if (StringUtils.isNotBlank(key) && null != properties && null != newValue) {
                properties.put(key, newValue);
            }
        }
    }

    public static IQtProducer getQtProducer() {
        DelegateKafkaProducer instance = KafkaProducerClientHolder.instance;
        if (null == instance) {
            throw new RuntimeException("build kafka producer client exception");
        }
        return instance;
    }

}
